package com.cscec8b.ct.analysis.mapper;


import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;

import java.io.IOException;


/**
 * @Copyright: Shanghai Definesys Company.All rights reserved.
 * @Description: 数据分析的mapper
 * @author: chuhaitao
 * @since: 2019/2/15 19:15
 * @history: 1.2019/2/15 created by chuhaitao
 */
public class CtTextMapper extends TableMapper<Text, Text> {
    /**
     * ImmutableBytesWritable： bytes,
     * Result：cell【】
     *
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {

        String rowKey = Bytes.toString(key.get());

        // 0_13956643892_20190211_13935457872_30000_1
        // 13956643892_20190211       1   dutation  30000
        //13956643892_201902          1   dutation  30000
        //13956643892_2019            1   dutation  30000

        String[] values = rowKey.split("_");

        //      1338989898_2018       1,  通话时长
        String call1 = values[1];
        String call2 = values[3];
        String time = values[2];
        String duration = values[4];

        /*主叫*/
        //年
        context.write(new Text(call1 + "_" + time.substring(0,4)), new Text(duration));
        //月
        context.write(new Text(call1 + "_" + time.substring(0,6)), new Text(duration));
        //日
        context.write(new Text(call1 + "_" + time), new Text(duration));

        /*被叫*/
        //年
        context.write(new Text(call2+ "_" + time.substring(0,4)), new Text(duration));
        //月
        context.write(new Text(call2 + "_" + time.substring(0,6)), new Text(duration));
        //日
        context.write(new Text(call2 + "_" + time), new Text(duration));
    }
}
